package com.qingyun.edusysteam.utils;

import com.qingyun.edusysteam.service.IPdfService;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static com.qingyun.edusysteam.utils.RedisConstants.*;

/**
 * PDF专用 Redis Stream 消费者
 */
@Component
public class StreamMessageConsumer {

    private final RedisTemplate<String, Object> redisTemplate;
    private final IPdfService pdfService;

    // 单线程消费（可根据需要改为多线程）
    private final ExecutorService executorService = Executors.newSingleThreadExecutor();

    public StreamMessageConsumer(RedisTemplate<String, Object> redisTemplate, IPdfService pdfService) {
        this.redisTemplate = redisTemplate;
        this.pdfService = pdfService;
        initConsumerGroup();
        startConsuming();
    }

    private void initConsumerGroup() {
        try {
            redisTemplate.opsForStream().createGroup(
                    PDF_TASK_QUEUE,
                    ReadOffset.latest(),
                    PDF_TASK_GROUP
            );
        } catch (Exception ignore) {}
    }

    public void startConsuming() {
        executorService.submit(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream().read(
                            Consumer.from(PDF_TASK_GROUP, PDF_TASK_CONSUMER),
                            StreamReadOptions.empty().block(Duration.ofSeconds(2)).count(10),
                            StreamOffset.create(PDF_TASK_QUEUE, ReadOffset.lastConsumed())
                    );
                    if (records == null || records.isEmpty()) continue;

                    for (MapRecord<String, Object, Object> r : records) {
                        try {
                            Map<Object, Object> map = r.getValue();
                            Object fid = map.get(FIELD_FEEDBACK_ID);
                            if (fid != null) {
                                Long feedbackId = Long.valueOf(String.valueOf(fid));
                                pdfService.generateAndSaveFeedbackPdf(feedbackId);
                            }
                            redisTemplate.opsForStream().acknowledge(PDF_TASK_GROUP, r);
                        } catch (Exception ex) {
                            // 不ack，进入pending
                        }
                    }
                } catch (Exception e) {
                    try { Thread.sleep(500L); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); }
                }
            }
        });
    }
}
    